package com.example.demo.Util;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;

import com.example.demo.config.Constant;

public class ConnectionUtils {


    public static RestClient createConnector() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(Constant.ES_USER, Constant.ES_USERP_PASSWORD));
        return RestClient.builder(new HttpHost(Constant.ES_NOED, 9200, "http"))
                .setRequestConfigCallback(builder -> builder.setConnectTimeout(60000).setSocketTimeout(60000))
            .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                    return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                }
            }).build();
    }
    public static Connection createConnection() {
        try {
            Configuration conf = HBaseConfiguration.create();
            //hadoop conf
            conf.set("hadoop.security.authentication", "kerberos");
            conf.set("hadoop.security.authorization", "true");
            // HBase conf
            conf.set("hbase.zookeeper.quorum", Constant.HOSTS);
            conf.set("zookeeper.znode.parent", "/hbase-secure");
            conf.set("hbase.zookeeper.property.clientPort", "2181");
            conf.set("hbase.security.authentication", "kerberos");
            conf.set("hbase.security.authorization", "true");
            conf.set("hbase.master.kerberos.principal", "hbase/_HOST@INDATAB.COM");
            conf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@INDATAB.COM");
            // login
            System.setProperty("java.security.krb5.conf", Constant.KRB5);
            UserGroupInformation.setConfiguration(conf);
            UserGroupInformation.loginUserFromKeytab(Constant.PRINCIPAL, Constant.KEY_TAB);
            return ConnectionFactory.createConnection(conf);
        } catch (IOException ioe) {
            System.err.println("Failed getting hbase connection.");
            ioe.printStackTrace();
            return null;
        }

    }
}
